﻿using Newtonsoft.Json;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace MX.Core.RabbitMQClient
{
    public class MQHelper
    {
        public static void SendMessage(string url, string queueName, object message)
        {
            if (message == null && message is object)
                throw new ArgumentNullException("参数message不能为空而且必须为对象，请检查参数");
            CheckPars(url, queueName);
            IModelWrapper iModel = ConnectionPool.GetOrCreateChannel(url, queueName, queueName, "exchangeDealer");
            try
            {
                IBasicProperties props = iModel.Model.CreateBasicProperties();
                props.ContentType = "text/plain";
                props.DeliveryMode = 2;
                iModel.Model.BasicPublish("exchangeDealer", queueName, props, Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(message)));
            }
            finally
            {
                iModel.SetNotBusy();
            }

        }

        private static void CheckPars(string url, string queueName)
        {
            if (string.IsNullOrEmpty(url))
                throw new ArgumentNullException("参数url不能为空，请检查参数");
            if (string.IsNullOrEmpty(queueName))
                throw new ArgumentNullException("参数queueName不能为空，请检查参数");

        }

        /// <summary>
        /// 成功接收
        /// </summary>
        /// <param name="url"></param>
        /// <param name="queueName"></param>
        /// <param name="ack"></param>
        public static void Ack(string url, string queueName, ulong ack)
        {
            CheckPars(url, queueName);
            if (ack < 0) throw new NotSupportedException("ACK 值错误");
            var consumerWrapper = ConsumerWrapper.GetConsumerWrapper(url, queueName);
            try
            {
                consumerWrapper.Model.BasicAck(ack, false);
            }
            finally
            {
                consumerWrapper.ModelWrapper.SetNotBusy();
            }
        }


        /// <summary>
        /// 重新进入消息队列
        /// </summary>
        /// <param name="url"></param>
        /// <param name="queueName"></param>
        /// <param name="ack"></param>
        /// <returns></returns>
        public static void Reject(string url, string queueName, ulong ack)
        {
            CheckPars(url, queueName);
            if (ack < 0) throw new NotSupportedException("ACK 值错误");
            var consumerWrapper = ConsumerWrapper.GetConsumerWrapper(url, queueName);
            try
            {
                consumerWrapper.Model.BasicReject(ack, true);
            }
            finally
            {
                consumerWrapper.ModelWrapper.SetNotBusy();
            }
        }

        /// <summary>
        /// 需要手动ACK
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="url"></param>
        /// <param name="queueName"></param>
        /// <returns></returns>
        public static NoAckResult<T> DequeueNoAck<T>(string url, string queueName)
        {

            var dequeue = DoDequeue<T>(url, queueName, false);
            return new NoAckResult<T>()
            {
                Body = dequeue.message,
                DeliveryTag = dequeue.result.DeliveryTag
            }; 
        }



        public static T Dequeue<T>(string url, string queueName)
        {
            return DoDequeue<T>(url, queueName, true).message;
        }

        public static bool Dequeue<T>(string url, string queueName, int millisecondsTimeout, out T message)
        {
            message = DoDequeue<T>(url, queueName, true, millisecondsTimeout).message;
            return message != null;
        }



        private static (T message, BasicGetResult result) DoDequeue<T>(string url, string queueName, bool autoAck = false, int millisecondsTimeout = int.MaxValue)
        {
            CheckPars(url, queueName);
            BasicGetResult result = null;
            var task = Task.Run(() =>
            {
                var stopTime = DateTime.Now.AddMilliseconds(millisecondsTimeout);
                while (true)
                {
                    var consumerWrapper = ConsumerWrapper.GetConsumerWrapper(url, queueName);
                    try
                    {

                        var consumer = consumerWrapper.Consumer;
                        consumerWrapper.ModelWrapper.IsBusy = true;
                        result = consumer.Model.BasicGet(queueName, autoAck);
                        if (result != null)
                        {
                            return;
                        }
                        if (stopTime < DateTime.Now)
                        {
                            return;
                        }

                    }
                    finally
                    {
                        consumerWrapper.ModelWrapper.SetNotBusy();
                    }
                    Task.Delay(10);
                }
            });
            task.Wait();
            if (result != null)
            {
                byte[] bytes = result.Body;
                var message = JsonConvert.DeserializeObject<T>(Encoding.UTF8.GetString(bytes));
                return (message, result);
            }
            else
            {
                return (default(T), result);
            }

        }


    }
}
